package com.xbai.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.util.parsing.json.JSON

/**
  * 数据读取与保存
  * Json文件: 如果 JSON 文件中每一行就是一个 JSON 记录，那么可以通过将 JSON 文件当做文本文件来读取，然后利用相关的 JSON 库对每一条数据进行 JSON 解析。
  * 注意：使用 RDD 读取 JSON 文件处理很复杂，同时 SparkSQL 集成了很好的处理 JSON 文件的方式，所以应用中多是采用 SparkSQL 处理 JSON 文件。
  *
  * Sequence文件: SequenceFile 文件是 Hadoop 用来存储二进制形式的 key-value 对而设计的一种平面文件 (Flat File)。
  * Spark 有专门用来读取 SequenceFile 的接口。在 SparkContext 中，可以调用 sequenceFile[ keyClass, valueClass](path)。
  *
  * 对象文件: 对象文件是将对象序列化后保存的文件，采用 Java 的序列化机制。
  * 可以通过 objectFile[k,v](path) 函数接收一个路径，读取对象文件，返回对应的 RDD，也可以通过调用 saveAsObjectFile() 实现对对象文件的输出。
  * 因为是序列化所以要指定类型。
  *
  * @author xbai
  * @Date 2021/1/2
  */
object Spark_ReadAndSave {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("ReadAndSave")
    val sc = new SparkContext(conf)

    val json: RDD[String] = sc.textFile("in/user.json")
    val result: RDD[Option[Any]] = json.map(JSON.parseFull)
    result.collect().foreach(println)

    val rdd: RDD[(Int, Int)] = sc.parallelize(Array((1, 2), (3, 4), (5, 6)))
    //    rdd.saveAsSequenceFile("output")
    val seq: RDD[(Int, Int)] = sc.sequenceFile("output")
    seq.collect().foreach(println)

    rdd.saveAsObjectFile("out/object")
    val objFile: RDD[(Int, Int)] = sc.objectFile("out/object")
    objFile.collect().foreach(println)
    sc.stop()
  }
}
